package com.atguigu.flink.datastreamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.EventListener;

/**
 * Created by Smexy on 2022/12/14
 *
 *      把每种传感器的水位和写入ck
 *          t1:   s1,200,100
 *          t2:   s1,201,300
 *
 create table ws
 (   id String,
 ts UInt64,
 vc UInt32
 )engine=ReplacingMergeTree(ts)
 order by (id);


 JdbcSink.sink(
 sqlDmlStatement,                       // mandatory
 jdbcStatementBuilder,                  // mandatory
 jdbcExecutionOptions,                  // optional。 自己构造
 jdbcConnectionOptions                  // mandatory
 );
 */
public class Demo5_JDBCSink
{
    public static void main(String[] args) throws Exception {

        //String url = "jdbc:mysql://hadoop104:3306/220730?useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true";

          /*
              String sql,
            JdbcStatementBuilder<T> statementBuilder, : 构造sql语句，填充占位符
            JdbcConnectionOptions connectionOptions :  连接参数
         */
        SinkFunction<WaterSensor> sink = JdbcSink.sink(
            "insert into ws values(?,?,?)",
            new JdbcStatementBuilder<WaterSensor>()
            {
                @Override
                public void accept(PreparedStatement ps, WaterSensor value) throws SQLException {
                    ps.setString(1, value.getId());
                    ps.setLong(2, value.getTs());
                    ps.setInt(3, value.getVc());
                }
            },
            JdbcExecutionOptions.builder().withBatchSize(200).withBatchIntervalMs(500).withMaxRetries(3).build(),
            new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:clickhouse://hadoop102:8123/default")
                .build()
        );

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId)
            .sum("vc");


        ds.addSink(sink);
        ds.print();

        env.execute();


    }

}
